package com.qupeng.demo.kafka.kafkaspringbootstreams.demo;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.streams.JsonSerde;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentCase;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentProfile;
import com.qupeng.demo.kafka.kafkaspringbootstreams.dashboard.model.AgentStatistic;
import com.qupeng.demo.kafka.kafkaspringbootstreams.wordcount.WordsStatistic;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.WindowStore;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
public class DemoStreams {
//    @Bean
    public KStream<Windowed<String>, AgentStatistic> agentStatisticKStream(StreamsBuilder myKStreamBuilder) {
        KTable<String, AgentProfile> agentProfiles = myKStreamBuilder
                .table("agent-profile", Consumed.with(Serdes.String(), new JsonSerde(AgentProfile.class)), Materialized.<String, WordsStatistic, WindowStore<Bytes, byte[]>>as("agent-profile-state").withValueSerde(new JsonSerde(AgentProfile.class)));
        KStream<String, AgentCase> agentCases = myKStreamBuilder.stream("agent-case", Consumed.with(Serdes.String(), new JsonSerde(AgentCase.class)));
        KStream<String, CustomerComplaint> complaints = myKStreamBuilder.stream("customer-complaint", Consumed.with(Serdes.String(), new JsonSerde(CustomerComplaint.class)));
        KStream<String, CaseState> caseStates = agentCases
                .leftJoin(agentProfiles, (key, agentCase, profile) -> {
                    if (null != profile) {
                        return new CaseState(profile.getId(), profile.getName(), agentCase, null);
                    } else {
                        return new CaseState(profile.getId(), "", agentCase, null);
                    }
                }).map((key, value) -> new KeyValue<>(value.getAgentCase().getId(), value));
        agentCases.print(Printed.toSysOut());
        KStream<String, CaseState> caseStatesWhitComplaints = caseStates
                .leftJoin(complaints, (key, caseState, complaint) -> {
                            if (null != complaint) {
                                caseState.setCustomerComplaint(complaint);
                            }
                            return caseState;
                        }, JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)).before(Duration.ofSeconds(0)),
                        StreamJoined.with(Serdes.String(), new JsonSerde<>(CaseState.class), new JsonSerde<>(CustomerComplaint.class)))
                .map(((key, value) -> new KeyValue<>(value.getAgentId(), value)));
        caseStates.print(Printed.toSysOut());
        KStream<Windowed<String>, AgentStatistic> caseStatistics = caseStatesWhitComplaints
                .groupByKey(Grouped.with(Serdes.String(), new JsonSerde<>(CaseState.class)))
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)).advanceBy(Duration.ofSeconds(300)))
                .aggregate(() -> new AgentStatistic(),
                        (key, value, agentStatistic) -> {
                            agentStatistic.setAgentId(value.getAgentId());
                            agentStatistic.setAgentName(value.getAgentName());
                            switch(value.getAgentCase().getStatus()) {
                                case "HANDLING" -> agentStatistic.getCaseHandling().add(value.getAgentCase().getId());
                                case "CLOSED" -> agentStatistic.getCaseHandled().add(value.getAgentCase().getId());
                                default -> {}
                            }
                            if (null != value.getCustomerComplaint()) {
//                                agentStatistic.setComplaintsAmount(agentStatistic.getComplaintsAmount() + 1);
                            }
                            return agentStatistic;
                        },
                        Materialized.<String, WordsStatistic, WindowStore<Bytes, byte[]>>as("agent-aggregates").withKeySerde(Serdes.String()).withValueSerde(new JsonSerde(AgentStatistic.class)))
                .toStream();
        caseStatistics.print(Printed.toSysOut());
        caseStatistics.to("agent-statistics", Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class, 5*60*1000L)));
        return caseStatistics;
    }

    public KStream<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
        KStream<Integer, String> stream = kStreamBuilder.stream("streamingTopic1");
        stream.mapValues((ValueMapper<String, String>) String::toUpperCase)
                .groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofMillis(1_000)))
                .reduce((String value1, String value2) -> value1 + value2,
                        Named.as("windowStore"))
                .toStream()
                .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
                .filter((i, s) -> s.length() > 40)
                .to("streamingTopic2");

        stream.print(Printed.toSysOut());

        return stream;
    }
}
